package com.rem.kafka.consumer;

import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.Logger;
import ch.qos.logback.classic.LoggerContext;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * 自定义kafka消费者
 *
 * @author Rem
 * @date 2023-01-03
 */
public class CustomerConsumer {
    static {
        LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory();
        Logger root = loggerContext.getLogger("root");
        root.setLevel(Level.INFO);
    }

    public static void main(String[] args) {
        Properties properties = new Properties();
        //连接kafka
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9093,localhost:9094,localhost:9095");
        //反序列化
//        properties.put(ConsumerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
//        properties.put(ConsumerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        //消费组id
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group01");
        //默认值为 true，消费者会自动周期性地向服务器提交偏移量
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        //如果设置了 enable.auto.commit 的值为 true 则该值定义了消费者偏移量向 Kafka 提交的频率，默认 5s
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);
        //当 Kafka 中没有初始偏移量或当前偏移量在服务器中不存在  （如，数据被删除了），该如何处理？
        // earliest：自动重置偏移量到最早的偏移量。
        // latest：默认，自动重置偏移量为最新的偏移量。
        // none：如果消费组原来的（previous）偏移量不存在，则向消费者抛异常。
        // anything：向消费者抛异常。
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        //Kafka 消费者和 coordinator 之间的心跳时间，默认 3s。该条目的值必须小于 session.timeout.ms ，也不应该高于session.timeout.ms 的 1/3。
        properties.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 3000);
        //Kafka 消费者和 coordinator 之间连接超时时间，默认 45s。超过该值，该消费者被移除，消费者组执行再平衡
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 45000);
        //消费者处理消息的最大时长，默认是 5 分钟。超过该值，该消费者被移除，消费者组执行再平衡
        properties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 5 * 60 * 1000);
        //默认 1 个字节。消费者获取服务器端一批消息最小的字节数
        properties.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1);
        //默认 500ms。如果没有从服务器端获取到一批数据的最小字节数。该时间到，仍然会返回数据。
        properties.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500);
        //默认 Default: 52428800（50 m）。消费者获取服务器端一批消息最大的字节数。如果服务器端一批次的数据大于该值（50m）仍然可以拉取回来这批数据，因此，这不是一个绝对最大值。
        // 一批次的大小受 message.max.bytes （brokerconfig）or max.message.bytes （topic config）影响。
        properties.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 52428800);

        //创建消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties, new StringDeserializer(), new StringDeserializer());

        /**
         * 订阅给定的主题列表以获取动态分配的分区。主题订阅不是增量的。此列表将替换当前分配（如果有的话）。
         * 无法通过assign(Collection)将主题订阅与组管理与手动分区分配相结合。如果给定的主题列表为空，则将其视为与unsubscribe()相同。
         */
        kafkaConsumer.subscribe(Collections.singletonList("first"));

        //消费数据
        while (true) {
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
            consumerRecords.forEach(System.err::println);
        }
    }
}
